package zzjson.rabbitmq.ack;

import com.rabbitmq.client.*;
import zzjson.rabbitmq.BaseRabbitMq;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * <p>****************************************************************************</p>
 * <p><b>Copyright © 2010-2019 shuncom team All Rights Reserved<b></p>
 * <ul style="margin:15px;">
 * <li>Description : zzjson.rabbitmq.ack</li>
 * <li>Version     : 1.0.0</li>
 * <li>Creation    : 2019年04月29日</li>
 * <li>@author     : zzy0_0</li>
 * </ul>
 * <p>****************************************************************************</p>
 */
public class ACKConsumer {
	private final static String QUEUE_NAME = "TEST_ACK_QUEUE";

	public static void main(String[] args) throws IOException {
		Connection conn = BaseRabbitMq.getConn();
		Channel channel = conn.createChannel();

		channel.queueDeclare(QUEUE_NAME, false, false, false, null);

		System.out.println("Waiting for message... ");

		DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
				String msg = new String(body, StandardCharsets.UTF_8);
				System.out.println("Received message :''" + msg + "");
				if (msg.contains("拒收")) {
					//	拒绝消息
					// requeue：是否重新入队列，true：是；false：直接丢弃，相当于告诉队列可以直接删除掉

					// TODO 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费
					channel.basicReject(envelope.getDeliveryTag(), false);
				} else if (msg.contains("异常")) {
					//	批量拒绝
					//	requeue:是否重新进入队列
					//	todo 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费
					channel.basicNack(envelope.getDeliveryTag(), true, false);
				} else {
					//手动应答
					// 如果不应答，队列中的消息会一直存在，重新连接的时候会重复消费
					channel.basicAck(envelope.getDeliveryTag(), true);
				}
			}
		};
		//开始获取消息，这里开启了手动应答
		// String queue, boolean autoAck, Consumer callback
		channel.basicConsume(QUEUE_NAME, false, consumer);
	}
}